package cn.nawang.ebeim.client.transfer;

import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE;

import java.io.File;
import java.io.FileInputStream;
import java.io.RandomAccessFile;
import java.net.URI;
import java.util.concurrent.TimeUnit;

import javax.activation.MimetypesFileTypeMap;

import org.apache.commons.codec.digest.DigestUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.nawang.ebeim.client.bean.ClientConfiguration;
import cn.nawang.ebeim.client.util.UploadProcessBarList;
import cn.nawang.ebeim.common.entity.FileInfo;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.ChannelProgressiveFuture;
import io.netty.channel.ChannelProgressiveFutureListener;
import io.netty.channel.DefaultFileRegion;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpRequest;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpObject;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpRequestEncoder;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseDecoder;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.handler.codec.http.QueryStringEncoder;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.handler.traffic.ChannelTrafficShapingHandler;

public class HttpUploadClient {
  private static final Logger LOG = LoggerFactory.getLogger(HttpUploadClient.class);
  private String host;
  private int port;
  private FileInfo info;
  private String dsId;
  private Long uploadLength;
  private Long preLength = 0l;
  private static final int connLine = 1;
  private static final int WRITE_WAIT_SECONDS = 5; // 多长时间未请求后，发送心跳
  
  private int flag = 0;// 如果为0：未建立连接；如果为1：建立成功;如果为2：传输失败；如果为3：传输成功
  private String md5;
  
  public HttpUploadClient(String host, int port) {
    this.host = host;
    this.port = port;
  }

  public int run(FileInfo info, long uploadLength,String md5) throws Exception {
    this.info = info;
    this.uploadLength = uploadLength;
    this.md5=md5;
    EventLoopGroup group = new NioEventLoopGroup(connLine);
    try {
      Bootstrap b = new Bootstrap();
      b.group(group).channel(NioSocketChannel.class);
      b.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
          ChannelPipeline pipeline = ch.pipeline();
          pipeline.addLast("decoder", new HttpResponseDecoder());
          pipeline.addLast("encoder", new HttpRequestEncoder());
          pipeline.addLast("http-encoder", new ChunkedWriteHandler());
          ch.pipeline().addLast("GLOBAL_TRAFFIC_SHAPING",
              new ChannelTrafficShapingHandler(ClientConfiguration.getWriteLimit(),
                  ClientConfiguration.getReadLimit(), ClientConfiguration.getCheckInterval()));
          pipeline.addLast("ping",
              new IdleStateHandler(0, WRITE_WAIT_SECONDS, 0, TimeUnit.SECONDS));
          pipeline.addLast("dispatcher", new HttpUploadClientInboundHandler());
        }
      });
      ChannelFuture f = b.connect(host, port).sync();
      f.channel().closeFuture().sync();
    } finally {
      LOG.info("退出" + flag);
      if (flag == 0) {
        LOG.info("网络信号差，未建立连接！文件(" + info.getPath() + ")连接不到[" + host + ":" + port + "]-----");
      }
      group.shutdownGracefully();
    }
    return flag;
  }

  private class HttpUploadClientInboundHandler extends SimpleChannelInboundHandler<HttpObject> {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
      flag = 1;
      File file = new File(info.getPath());
      if (!file.exists()) {
        LOG.info("不存在文件：" + file.getAbsolutePath());
        flag = 2;
      } else {
        final RandomAccessFile raf = new RandomAccessFile(file, "r");
        long fileLength = raf.length();
        QueryStringEncoder encoder = new QueryStringEncoder("upload");
        encoder.addParam("type", String.valueOf(info.getType()));
        encoder.addParam("version", info.getVersion());
        encoder.addParam("dataId", info.getDataId());
        encoder.addParam("signature", info.getSignature());
        encoder.addParam("md5", md5);
        URI uriGet = new URI(encoder.toString());
        HttpRequest request =
            new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, uriGet.toASCIIString());
        long count = fileLength - uploadLength;
        HttpHeaders.setContentLength(request, count);
        MimetypesFileTypeMap mimeTypesMap = new MimetypesFileTypeMap();
        request.headers().set(CONTENT_TYPE, mimeTypesMap.getContentType(file.getPath()));
        if (HttpHeaders.isKeepAlive(request)) {
          request.headers().set("CONNECTION", HttpHeaders.Values.KEEP_ALIVE);
        }
        dsId = info.getDataId();
        ctx.write(request);
        LOG.info(info.getDataId()+":文件签名：【"+info.getSignature()+"】版本号【"+info.getVersion()+"】--》["+host+"]开始传输");
        ChannelFuture sendFileFuture;
        sendFileFuture = ctx.write(new DefaultFileRegion(raf.getChannel(), uploadLength, count),
            ctx.newProgressivePromise());
        sendFileFuture.addListener(new ChannelProgressiveFutureListener() {
          @Override
          public void operationProgressed(ChannelProgressiveFuture future, long progress,
              long total) {
            if (total < 0) {
              LOG.debug(future.channel() + " Transfer progress: " + progress);
            } else {
              LOG.debug(future.channel() + " Transfer progress: " + progress + " / " + total);
            }
            if (info.getType() == FileInfo.TYPE_FILE_UPLOAD) {
              UploadProcessBarList.summationTotal(dsId, progress - preLength);
              preLength = progress;
            }
          }

          @Override
          public void operationComplete(ChannelProgressiveFuture future) throws Exception {
            raf.close();
            if (!future.isSuccess()) {
              LOG.info(future.channel() + "传输到服务端失败");
            } else {
              LOG.info(future.channel() + "传输到服务端完成");
            }
          }
        });

        ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
      }

    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
      if (msg instanceof HttpResponse) {
        HttpResponse response = (HttpResponse) msg;
        LOG.info("文件：(" + info.getPath() + ")--STATUS: " + response.getStatus());
        if (!response.getStatus().equals(HttpResponseStatus.OK)) {
          flag = 2;
          LOG.info("error: " + response.headers().get("error"));
        } else {
          flag = 3;
        }
      }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
      ctx.close();
    }
  }
}

